package com.exd.kafka.task;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Base64;
import java.util.Collections;
import java.util.List;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.exd.kafka.service.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class ReadDataTask {
	
	@Autowired
	private ProducerService producerService;
	@Autowired
	private Environment environment;
	
//	@Scheduled(cron = "0 0/3 * * * ? ")
    public void readDataExecuteTaskJob() throws Exception {
		String kafkaTopic = environment.getProperty("spring.kafka.producer.topic-name");
		String path = environment.getProperty("spring.kafka.producer.properties.exd.source.path");

		int second = 130;
		int cycle = 10;
		int interval = 0;
		Long milliSecond = MockVdataVsw.getLocalDateTimeSecond(LocalDateTime.now()) - second;
		String data = MockVdataVsw.mockVswString(path, milliSecond, second, cycle, interval);
		String[] vins = "LDP91C968RE000187".split(",");

		for (int i = 0; i < vins.length; i++) {
			JSONObject jsonObject = new JSONObject();
			jsonObject.put("collectionTime", milliSecond * 1000);
			jsonObject.put("data", data);
			jsonObject.put("vin", vins[i]);
			producerService.sendMessageToKafka(kafkaTopic, jsonObject.toJSONString());
			System.out.println(jsonObject.toJSONString());
		}
	}

//	@Scheduled(cron = "0 0/1 * * * ? ")
	public void readDataExecuteTaskJob2() throws Exception {
		String kafkaTopic = environment.getProperty("spring.kafka.producer.topic-name");
		String path = environment.getProperty("spring.kafka.producer.properties.exd.source.path");
		int second = 60;
		int cycle = 10;
		Long milliSecond = MockVdataVsw2.getLocalDateTimeSecond(LocalDateTime.now()) - second;
		String data = MockVdataVsw2.mockVswString(path, milliSecond, second, cycle,"msg_time,Viu_mr.1487.OIB_VCU_Test2.HvStateNr,Viu_r.484.VIU_OIB_EBS_0x35_VIU_R.ebs_SOC_STATE_60_VIUR,Viu_r.481.VIU_OIB_EBS_0x34_VIU_R.ebs_T_BATT_60_VIUR,Viu_r.481.VIU_OIB_EBS_0x34_VIU_R.ebs_I_RANGE_60_VIUR,Viu_r.481.VIU_OIB_EBS_0x34_VIU_R.ebs_I_BATT_60_VIUR,Viu_r.481.VIU_OIB_EBS_0x34_VIU_R.ebs_U_BATT_60_VIUR,Viu_r.484.VIU_OIB_EBS_0x35_VIU_R.ebs_SOC_60_VIUR");

		String[] vins = "LDP91H964RE000135".split(",");
		for (int i = 0; i < vins.length; i++) {
			JSONObject jsonObject = new JSONObject();
			jsonObject.put("collectionTime", milliSecond * 1000);
			jsonObject.put("data", data);
			jsonObject.put("vin", vins[i]);
			producerService.sendMessageToKafka(kafkaTopic, jsonObject.toJSONString());
			System.out.println(jsonObject.toJSONString());
		}
	}

//	@Scheduled(cron = "0 0/7 * * * ? ")
	public void readDataExecuteTaskJob3() throws Exception {
		String kafkaTopic = environment.getProperty("spring.kafka.producer.topic-name");
		String path = environment.getProperty("spring.kafka.producer.properties.exd.source.path");

		int second = 420;
		int cycle = 10;
		Long milliSecond = MockVdataVsw3.getLocalDateTimeSecond(LocalDateTime.now()) - second;
		String data = MockVdataVsw3.mockVswString(path, milliSecond, second, cycle);

		String[] vins = "LGB32XE81NW801645".split(",");

		for (int i = 0; i < vins.length; i++) {
			JSONObject jsonObject = new JSONObject();
			jsonObject.put("collectionTime", milliSecond * 1000);
			jsonObject.put("data", data);
			jsonObject.put("vin", vins[i]);
			producerService.sendMessageToKafka(kafkaTopic, jsonObject.toJSONString());
			System.out.println(jsonObject.toJSONString());
		}
	}

	@Scheduled(cron = "0 0/12 * * * ? ")
	public void readDataExecuteTaskJob4() throws Exception {
		String kafkaTopic = environment.getProperty("spring.kafka.producer.topic-name");
		String path = environment.getProperty("spring.kafka.producer.properties.exd.csv.path");

		int second = 720;
		int cycle = 1;
		Long milliSecond = (MockVdataCsvToVsw.getLocalDateTimeSecond(LocalDateTime.now()) - second) * 1000;

		String writePath =  path + milliSecond +"/";
		File writefile = new File(writePath);
		if (!writefile.exists()) {
			writefile.mkdirs();
		}

		Long time = milliSecond;

		List<String> fileList = Arrays.asList("0", "1", "2", "3", "4", "5");

		File[] files = new File(path).listFiles();
		for (File file : files) {
			if (file.getPath().endsWith(".csv")){
				for (String name : fileList) {
					if (file.getName().contains(name)){
						String outPath = writePath + "alarm"+name + ".csv";
						String vswPath = writePath + name + ".vsw";
						File outFile = new File(outPath);
						time = MockVdataCsvToVsw.writeCsvFile(file, outFile, time, cycle);
						MockVdataCsvToVsw.csvToVswFile(outPath, vswPath);

						byte[] vswBytes = MockVdataCsvToVsw.readFromByteFile(vswPath);
						String vswData = Base64.getEncoder().encodeToString(vswBytes);

						JSONObject jsonObject = new JSONObject();
						jsonObject.put("collectionTime", milliSecond);
						jsonObject.put("data", vswData);
						jsonObject.put("vin", "LGB32XE81NW801645");
						System.out.println(jsonObject.toJSONString());

						producerService.sendMessageToKafka(kafkaTopic, jsonObject.toJSONString());
					}
				}
			}
		}
	}
}
